package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SparkApiTest_TOPN {


  implicit val myOrder = new Ordering[(Int, Int)] {
    override def compare(x: (Int, Int), y: (Int, Int)) = y._2.compareTo(x._2)
  }


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("topN").setMaster("local")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    val file = context.textFile("test_file/topnfile")
    //    样例数据：1970-8-8	32

    /**
     * 1、一次shuffle
     * ①、先根据年月进行分组
     * ②、然后对每一组（同年同月不同日）的数据中同一天的数据进行去重（只保留最大值的一天）
     * ③、对每一组（同年同月不同日）的数据根据温度进行排序
     *
     *
     * 问题比较明显：容易造成OOM，在去重阶段，需要把数据写入一个map，如果map值非常多，则会内存溢出
     */
    //    val maped: RDD[((Int, Int), (Int, Int))] = file.map(line => {
    //      val strings = line.split("\t")
    //      val ymdStrs = strings(0).split("-")
    //      ((ymdStrs(0).toInt, ymdStrs(1).toInt), (ymdStrs(2).toInt, strings(1).toInt))
    //    })
    //
    //
    //    val grouped: RDD[((Int, Int), Iterable[(Int, Int)])] = maped.groupByKey() //根据年月分组
    //
    //    val resultRdd: RDD[((Int, Int), List[(Int, Int)])] = grouped.mapValues(values => {

    //      val map = new mutable.HashMap[Int, Int]()
    //      values.foreach(item => if (map.getOrElse(item._1, 0) < item._2) map.put(item._1, item._2))
    //      map.toList.sorted //默认倒序
    //    })
    //    resultRdd.foreach(println)

    /**
     * 2、2次shuffle
     * 典型的时间换空间
     * 与1比，提前多做1次reduce来首先完成同一天的去重,这样在groupbykey的时候，数量一定会比1方法中要少许多  但实际上也可能存在OOM的风险
     */
    //    val mapped: RDD[((Int, Int, Int), Int)] = file.map(line => {
    //      val strings = line.split("\t")
    //      val ymdStrs = strings(0).split("-")
    //      ((ymdStrs(0).toInt, ymdStrs(1).toInt, ymdStrs(2).toInt), (strings(1).toInt))
    //    })
    //    val maxNumOneDayDatas: RDD[((Int, Int, Int), Int)] = mapped.reduceByKey((oldV, newV) => if (newV > oldV) newV else oldV)
    //    //只留同一天里最大的温度的那天
    //    val preRDD: RDD[((Int, Int), (Int, Int))] = maxNumOneDayDatas.map(item => ((item._1._1, item._1._2), (item._1._3, item._2)))
    //    val groupedRDD: RDD[((Int, Int), Iterable[(Int, Int)])] = preRDD.groupByKey()
    //    val resultRDD: RDD[((Int, Int), List[(Int, Int)])] = groupedRDD.mapValues(items => items.toList.sorted)
    //    resultRDD.foreach(println)


    /**
     * 3、先全排序，再去重，再分组
     * 有问题！！！　顺序不对。。。　　　为什么？？　　
     * 第一次相当是根据年、月、温度　进行分区
     * 第二次根据 年、月、日 进行分区
     *
     * 每一次shuffle都会对数据进行重新分区计算
     * 两次不同维度的分区标准会导致数据可能产生错乱，去到不同的位置，导致第一次的全排序失效
     * 全排序似乎需要在一次shuffle内完成的才可以保证？？
     *
     */
    //    有问题的版本:
    //    val mapped = file.map(line => {
    //      val strings = line.split("\t")
    //      val ymdStrs = strings(0).split("-")
    //      (ymdStrs(0).toInt, ymdStrs(1).toInt, ymdStrs(2).toInt, strings(1).toInt)
    //    })
    //    //根据年月温度全排序
    //    val sorted: RDD[(Int, Int, Int, Int)] = mapped.sortBy(item => (item._1, item._2, item._4), false)
    //    //同一天去重，保留最大值的天数
    //    val reduced: RDD[((Int, Int, Int), Int)] = sorted.map(item => ((item._1, item._2, item._3), item._4)).reduceByKey((oldV, newV) => if (newV > oldV) newV else oldV)
    //    //根据年月分组
    //    val grouped: RDD[((Int, Int), Iterable[(Int, Int)])] = reduced.map(item => ((item._1._1, item._1._2), (item._1._3, item._2))).groupByKey()
    //    grouped.foreach(println)

    /**
     * 其实也不完全，多级shuffle  后续shuffle参与分区的key一定得是前置分区 key的子集  ，就可以保证顺序性
     * 4、先使用年月温度进行全排序，再根据年月进行分组
     * 分组里处理同一天重复问题
     * 最后的结果没有问题
     * 但是，依然有groupByKey这个操作，就意味着，依然会存在OOM风险
     */

    // 正确案例：
    //    val sorted: RDD[(Int, Int, Int, Int)] = mapped.sortBy(t4 => (t4._1, t4._2, t4._4), false)
    //    //    val reduced: RDD[((Int, Int, Int), Int)] = sorted.map(t4 => ((t4._1, t4._2, t4._3), t4._4)).reduceByKey((x: Int, y: Int) => if (y > x) y else x)
    //    val maped = sorted.map(t4 => ((t4._1, t4._2), (t4._3, t4._4)))
    //    val grouped = maped.groupByKey()
    //
    //    grouped.mapValues(values => {
    //      val map = new mutable.HashMap[Int, Int]()
    //      values.foreach(item => if (map.getOrElse(item._1, 0) < item._2) map.put(item._1, item._2))
    //      map.toList
    //    }).foreach(println)

    /**
     * 5、分布式计算的核心思想：调优天下无敌：自定义的combineByKey
     * 此方案完美解决OOM
     * 想一下：如果能在map端 combine的时候就可以实现排序，去重就好了。。
     * 自定义的combineByKey 尽量压缩积压在内存的数据量在最小
     *
     */
    val mapped: RDD[((Int, Int), (Int, Int))] = file.map(line => {
      val strings = line.split("\t")
      val ymdStrs = strings(0).split("-")
      ((ymdStrs(0).toInt, ymdStrs(1).toInt), (ymdStrs(2).toInt, strings(1).toInt))
    })

    val resultRes: RDD[((Int, Int), Array[(Int, Int)])] = mapped.combineByKey( //同年同月
      //如果一个key有100个value，我用3个宽度的数组，每次排序排3个，每次从100个value拿一个进来，这样，内存是最少占用
      (item: (Int, Int)) => Array(item, (0, 0), (0, 0)), //待获取前N个最大温度值的天 则存放 N+1个， 最后一个作为每次value进来时的缓冲存放区
      (oldV: Array[(Int, Int)], newV: (Int, Int)) => {
        println(s"oldArr:${oldV.toList}, newArr:$newV")
        var flag = 0
        for (i <- oldV.indices) {
          if (newV._1 == oldV(i)._1) { //同一天
            if (newV._2 > oldV(i)._2) {
              oldV(i) = newV
            }
            flag = 1
          }
        }
        if (flag == 0) {
          oldV(oldV.length - 1) = newV
        }
        //        oldV.sorted   每次都会生成一个新对象，有造成OOM的风险
        util.Sorting.quickSort(oldV) //对原有Array进行排序
        oldV
      },
      //     combine合并的时候也需要对同一天去重，例如一边是1号11度，1边是1号22度，应该是只留下1号22度，留下同一天中最高的温度
      (oldV: Array[(Int, Int)], newV: Array[(Int, Int)]) => {
        //            做法1：
        var addNums = Seq.newBuilder[(Int, Int)]
        for (newI <- newV.indices) {
          var flag = 0
          for (oldI <- oldV.indices) {
            if (oldV(oldI)._1 == newV(newI)._1) {
              if (oldV(oldI)._2 < newV(newI)._2) {
                oldV(oldI) = newV(newI)
              }
              flag = 1
            }
            if (flag == 0) {
              addNums.+=(newV(newI))
            }
          }
        }
        val result: Array[(Int, Int)] = oldV.union(addNums.result())
        util.Sorting.quickSort(result)
        result

        //            做法2：
        //            val tuples: Array[(Int, Int)] = oldV.union(newV)
        //            util.Sorting.quickSort(tuples)
        //            var index = 0
        //            var addNums = Seq.newBuilder[(Int, Int)]
        //            while (index < tuples.length) {
        //              if (index < tuples.length - 1 && tuples.length > 1) {
        //                if (tuples(index)._1 == tuples(index + 1)._1) {
        //                  if (tuples(index)._2 < tuples(index + 1)._2) {
        //                    addNums.+=(tuples(index + 1))
        //                  } else {
        //                    addNums.+=(tuples(index))
        //                  }
        //                  index += 2
        //                }
        //
        //              } else {
        //                index += 1
        //              }
        //            }
        //            addNums.result().toArray
      }
    )


    resultRes.mapValues(item => item.filter(one => one != (0, 0)).take(2))
      .map((item: ((Int, Int), Array[(Int, Int)])) => (item._1, item._2.toList)).foreach(println)

    //    val data: RDD[String] = context.parallelize(List(
    //      "hello world",
    //      "hello spark",
    //      "hello world",
    //      "hello hadoop",
    //      "hello world",
    //      "hello msb",
    //      "hello world"
    //    ))
    //    val words: RDD[String] = data.flatMap(_.split(" "))
    //    val kv: RDD[(String, Int)] = words.map((_, 1))
    //    val res: RDD[(String, Int)] = kv.reduceByKey(_ + _)
    //    //key没有发生变化，分区器没有发生变化，分区数没有发生变化，且是kv的，
    //    // 那么建议mapValues  flatMapValues这种，
    //    //和map的区别是，map会丢掉原有的分区器，进行重新的shuffle操作，而mapValues是继续使用原有的
    //    // 因为上述key等没有变化，那么其实没必要再次进行一次shuffle，
    //    // 多进行的这一次shuffle因为key等没有变化，会依旧拉回本机，但多了shuffle写，shuffle读，落地文件的过程，但实际没有意义
    //    // 完全可以利用现有数据进行处理，这样进而是一种利用了spark的优化
    //    val res01: RDD[(String, Int)] = res.mapValues(x => x * 10)
    //    //    val res01: RDD[(String, Int)] = res.map(x => (x._1, x._2 * 10))
    //    val res02: RDD[(String, Iterable[Int])] = res01.groupByKey()
    //    res02.foreach(println)


    while (true) {

    }


  }

}
